package dshumsky.akka;

import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.routing.RoundRobinRouter;


public class MasterActor
    extends UntypedActor {

    private int mapActorMessage = 0;
    private int reduceActorMessage = 0;
    private int aggregateActorMessage = 0;

    private int needResultAfter = 0;
    private ActorRef needResultSender = null;
    private RoundRobinRouter router = new RoundRobinRouter(4);
    ActorRef mapActor = getContext().actorOf(new Props(MapActor.class).withRouter(router), "map");
    ActorRef reduceActor = getContext().actorOf(new Props(ReduceActor.class).withRouter(router), "reduce");
    ActorRef aggregateActor = getContext().actorOf(new Props(AggregateActor.class), "aggregate");

    @Override
    public void onReceive(Object message)
        throws Exception {

        if (message instanceof String) {
            mapActorMessage++;

            if ((mapActorMessage % 200) == 0) {
                System.out.println("m=" + mapActorMessage + ", r=" + reduceActorMessage + ", a=" + aggregateActorMessage);
            }

            mapActor.tell(message, getSelf());
        } else if (message instanceof MapData) {
            reduceActorMessage++;

            if ((reduceActorMessage % 10) == 0) {
                System.out.println("m=" + mapActorMessage + ", r=" + reduceActorMessage + ", a=" + aggregateActorMessage);
            }

            reduceActor.tell(message, getSelf());
        } else if (message instanceof ReduceData) {
            aggregateActorMessage++;

            if ((aggregateActorMessage % 10) == 0) {
                System.out.println("m=" + mapActorMessage + ", r=" + reduceActorMessage + ", a=" + aggregateActorMessage);
            }

            aggregateActor.tell(message, self());

            if (aggregateActorMessage == needResultAfter) {
                aggregateActor.tell(new Result(), needResultSender);
            }
        } else if (message instanceof Result) {
            needResultAfter = mapActorMessage;
            needResultSender = getSender();
        } else {
            unhandled(message);
        }
    }
} // end class MasterActor
